package com.wangt.biz.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.util.List;

@Slf4j
public class MqttConsumerCallBack implements MqttCallbackExtended {

    MqttClient client;
    List<String> topics;
    public MqttConsumerCallBack(MqttClient client, List<String> topics) {
        this.client = client;
        this.topics = topics;
    }

    /**
     * 客户端断开连接的回调
     */
    @Override
    public void connectionLost(Throwable throwable) {
        System.out.println("与服务器断开连接，可重连");
    }

    /**
     * 消息到达的回调
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println(String.format("接收消息主题 : %s",topic));
        System.out.println(String.format("接收消息Qos : %d",message.getQos()));
        System.out.println(String.format("接收消息内容 : %s",new String(message.getPayload())));
        System.out.println(String.format("接收消息retained : %b",message.isRetained()));
    }

    /**
     * 消息发布成功回调
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {}


    /**
     * 连接成功后 回调 自动订阅系统中设置的主题
     * @param reconnect If true, the connection was the result of automatic reconnect.
     * @param serverURI The server URI that the connection was made to.
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("==客户端连接成功==");
        try {
            log.info("==订阅Topic==");
            for (String topic : topics) {
                client.subscribe(topic,0);
            }
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }
}
